import { Connection, Channel, Options, Replies } from "amqplib";
import { useMq } from "./useMq";
import { QueueHandler } from "./types";
import { delay } from "./util";
/**
 * 从指定的队列中监听消息。
 * 
 * 此函数旨在建立与消息队列的连接，并从指定的队列中消费消息。如果在操作过程中发生错误，它将尝试关闭当前连接和通道，并重新尝试连接和监听。
 * 这种设计确保了在遇到临时故障时，能够自动恢复监听，保持系统的高可用性。
 * 
 * @param queueName 队列的名称，用于指定需要监听的消息队列。
 * @param handler 消息处理程序，当收到消息时将调用此函数处理消息。
 * @param options 队列的配置选项，用于在AMQP中配置队列的特性。
 */
export async function listenFromQueue(queueName: string, handler: QueueHandler, options: Options.AssertQueue, count: number = 20) {
    let con: Connection | undefined = undefined;
    let channel: Channel | undefined = undefined;
    let consume: Replies.Consume | undefined = undefined;
    try {
        // 获取消息队列的连接。
        con = await useMq()
        // 创建一个消息通道。
        channel = await con.createChannel()
        // 断言队列的存在，如果队列不存在则创建它。
        await channel.assertQueue(queueName, options).catch(e => console.error(e))
        // 开始消费队列中的消息，每收到一条消息就调用处理程序进行处理。
        await channel.prefetch(count)
        console.log(`channel prefetch: ${count}`)
        consume = await channel.consume(queueName, (msg) => {
            if (msg) handler(msg, channel!).catch(e => console.error(e))
        }, { noAck: false });
        return async () => {
            if (channel) {
                if (consume) {
                    await channel.cancel(consume.consumerTag).catch(e => console.error(e));
                }
                await channel?.close().catch(e => console.error(e))
            }
            if (con) {
                await con?.close().catch(e => console.error(e))
            }
        }
    } catch (e) {
        console.error(e)
        if (channel) {
            if (consume) {
                await channel.cancel(consume.consumerTag).catch(e => console.error(e));
            }
            await channel?.close().catch(e => console.error(e))
        }
        if (con) {
            await con?.close().catch(e => console.error(e))
        }
        throw e;
    }
}